个人笔记

外观

2026-11-09
1520
5.9m

iceoryx2 事件机制：一次性通知与数据控制流分离 ​

发布订阅解决「传什么数据」，但很多场景只需要一个信号：电量低了、撞墙了、传感器掉了。用 pub-sub 传这类信号，要么发空消息、要么让接收方高频轮询，都是绕路。iceoryx2 的事件（Event）机制专为「通知状态变化」设计：一次性信号、不排队、不丢。本文介绍事件模型的核心语义、C++ 双端实现，以及它与发布订阅组合出的「数据流与控制流分离」模式，适合正在使用或评估 iceoryx2 的开发者。

状态与事件 ​

先分清两个概念：

  • 状态（State）：内存中的当前事实，例如电池当前是低电量还是正常；
  • 事件（Event）：状态发生了变化这一通知本身。

事件机制由两种端口组成：通知器（Notifier） 发送通知，监听器（Listener） 等待通知。监听器平时休眠，只在真正需要关注的事件发生时被唤醒。

核心语义：不排队、会合并、顺序未定义 ​

这是理解事件机制最关键的部分，它和消息队列是两种模型：

  • 不排队：事件不像 pub-sub 消息那样在缓冲区里排队等待消费。每个事件 ID 更接近一个带锁存的状态标志位；
  • 会合并：如果通知器在监听器处理之前连续发送了多次同一事件，监听器醒来只会感知到一次「该状态变过」；
  • 顺序未定义：多个事件到达的先后顺序没有保证，监听器只知道「至少有某状态变化发生过」；
  • 不丢失：已发出的事件在监听器观察到之前不会丢失——综上，语义是「每个事件 ID 至少被通知一次，可见次数至多一次」。

这套语义适合「电平触发」型条件：只关心当前事实（电池是否低），不关心它被报了几次。如果需要每条消息都独立送达，那应该用 pub-sub 而不是事件。

双端实现 ​

通知器端 ​

创建节点与事件服务，用 EventId 区分不同类型的事件：

cpp
#include "iox2/iceoryx2.hpp"

using namespace iox2;

auto node = NodeBuilder().create<ServiceType::Ipc>().value();

auto service = node.service_builder(ServiceName::create("system_health_events").value())
                   .event()
                   .open_or_create()
                   .value();

auto notifier = service.notifier_builder().create().value();

// 为不同类型的事件分配唯一标识
const auto wall_was_hit = EventId(0);
const auto battery_is_low = EventId(1);

while (node.wait(iox2::bb::Duration::from_secs(1)).has_value()) {
    if ( bump_sensor_was_activated() ) {
        notifier.notify_with_custom_event_id(wall_was_hit).value();
    }

    if ( battery_state() < battery_threshold ) {
        notifier.notify_with_custom_event_id(battery_is_low).value();
    }
}

监听器端 ​

打开同一个事件服务，阻塞等待并在回调里按事件 ID 分发处理：

cpp
#include "iox2/iceoryx2.hpp"

using namespace iox2;

auto node = NodeBuilder().create<ServiceType::Ipc>().value();

auto service = node.service_builder(ServiceName::create("system_health_events").value())
                   .event()
                   .open_or_create()
                   .value();
auto listener = service.listener_builder().create().value();

while (listener.blocking_wait(
    [](auto event) {
        if ( event.id() == battery_is_low ) {
            activate_battery_warning_light();
        }

        if ( event.id() == wall_was_hit ) {
            go_into_parking_position();
        }
    }).has_value()) {}

注意 blocking_wait 的回调参数是事件对象，取标识用 event.id()。

事件驱动模式：数据流与控制流分离 ​

单一模式之外，iceoryx2 官方示例展示了一个更有价值的组合思路：

  • 数据流描述「函数接收什么输入」——由发布订阅、请求响应承载；
  • 控制流描述「函数何时执行」——由事件承载。

两者分离后，不需要用「发空消息」或「过早唤醒」这类 hack 来触发执行，还能支持无数据唤醒和多输入等待（所有输入到齐前不中断）。

以机器人紧急刹车系统为例：

  • 选择性响应：只在障碍物接近时刹车，而不是每个数据样本都触发一轮处理；
  • 趋势分析：刹车决策需要最近 3 个距离样本（位置、速度、加速度）；
  • 容错：传感器死亡时进入安全状态（停车）；刹车后重新启动时仍能取到历史数据。

发布者：pub-sub 传数据 + 事件传信号 ​

cpp
auto node = NodeBuilder().create<ServiceType::Ipc>().value();

// 1. 发布订阅服务（距离数据流）
auto pubsub_service = node.service_builder(
                        ServiceName::create("distance_to_obstacle").value())
                   .publish_subscribe<Distance>()
                   .subscriber_max_buffer_size(3)
                   .history_size(3)
                   .subscriber_max_borrowed_samples(3)
                   .open_or_create()
                   .value();

// 2. 事件服务（控制信号）
auto ultra_sonic_service_dead = EventId(10);
auto event_service = node.service_builder(
                        ServiceName::create("distance_to_obstacle").value())
                   .event()
                   .notifier_dead_event(ultra_sonic_service_dead)
                   .open_or_create()
                   .value();

auto publisher = pubsub_service.publisher_builder().create().value();
auto notifier = event_service.notifier_builder().create().value();
auto obstacle_too_close = EventId(5);

while (node.wait(iox2::bb::Duration::from_millis(100)).has_value()) {
    auto sample = publisher.loan_uninit().value();

    auto distance = get_ultra_sonic_sensor_distance();
    auto initialized_sample = sample.write_payload(Distance { distance, 42.0 });

    if (distance < distance_threshold) {
        notifier.notify_with_custom_event_id(obstacle_too_close).value();
    }
}

几项配置的含义：

  • subscriber_max_buffer_size(3)：订阅者最大缓冲区大小；
  • history_size(3)：历史样本保留数量——新订阅者或重启的处理方能立即拿到最近 3 个样本；
  • subscriber_max_borrowed_samples(3)：同时可借用的样本数上限；
  • notifier_dead_event(ultra_sonic_service_dead)：把某个 EventId 注册为「通知方死亡」信号，依托 iceoryx2 内置的参与者健康监控自动触发。

订阅者：被事件唤醒，醒来再取数据 ​

订阅者不再按 100ms 轮询，而是阻塞在事件上。被唤醒后，如果是「障碍物过近」，一次性取走缓冲区里的最近 3 个样本做趋势分析；如果是「传感器死亡」，直接进安全状态：

cpp
while (listener.blocking_wait(
    [](auto event) {
        // 传感器死亡事件
        if ( event.id() == ultra_sonic_service_dead ) {
            go_into_parking_position();
        }

        // 障碍物过近事件
        if ( event.id() == obstacle_too_close ) {
            std::vector<Sample<ServiceType::Ipc, Distance, void>> last_samples;

            auto sample = subscriber.receive().value();
            while (sample.has_value()) {
                last_samples.push_back(std::move(sample.value()));
                if (last_samples.size() == 3) break;
                sample = subscriber.receive().value();
            }

            perform_break(last_samples);
        }
    }).has_value()) {}

数据与控制两条通路各司其职：pub-sub 负责持续的距离样本流并保留历史，事件负责「现在必须处理」的触发时机。

要点 ​

  • 事件是锁存的状态标志，不是消息队列：会合并、顺序未定义、每个事件 ID 至少通知一次
  • 需要每条消息独立送达时用 pub-sub，需要「状态变了叫我」时用事件
  • EventId 区分事件类型，blocking_wait 回调里用 event.id() 分发
  • 数据流与控制流分离是事件模式的正确打开方式：pub-sub 管数据与历史，事件管触发时机
  • notifier_dead_event 借助内置健康监控实现参与者死亡检测，容错场景直接可用
最近更新

基于 VitePress 构建

Copyright © 2026 ww1820